package node;

import java.io.Serializable;
import java.rmi.AccessException;
import java.rmi.AlreadyBoundException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import remote.RemoteDistributedHashMap;
import remote.UnicastRemoteDistributedHashMap;

import log.Logger;
import messaging.AddNodeMessage;
import messaging.InterNodeMessage;
import messaging.MessageHashing;
/**
 * This class aids in processing the messages received on the channels - 
 * for Add node, remove node and hash table operations.
 * @author Team4
 * @version Spring 2013
 *
 */
public class QueueHelper
{
  /**
   * Holds a reference to node manager.
   */
  private NodeManager nodeman;
  /**
   * Constructor.
   */
  public QueueHelper(NodeManager nm)
  {
    this.nodeman = nm;
  }
  /**
   * process add node message.
   * @param message
   * @param my_node_id
   */
  public synchronized void processAddNodeMessage(final InterNodeMessage my_message, 
                                                 final int my_node_id)
  {
    try
    {
      Logger.out(" inside node " + my_node_id);
      int destNode;
      String dest_str = null;
      if (my_message.isAnswer_set() && my_message.getMy_originaldest_node() == my_node_id)
      {
        Logger.out(" new node message received from predecessor and ready to add new node(s)");
        addVirtualNodes(my_message, my_node_id);
        final String unbind = new Integer(my_node_id).toString() + "V";
        NodeManager.registry.unbind(unbind);
        Logger.out(" list of updated nodes in registry are " +
            Arrays.toString(NodeManager.registry.list()));
      }
      if (!my_message.isAnswer_set() && my_message.getMy_originaldest_node() == my_node_id)
      {
      //message not processed and and it is at the right node to process.
        Logger.out(" inside the predecessor node - " + my_node_id);
        ArrayList<Integer> nodes = my_message.get_AddNode_Message().getAddNodesList();
        ArrayList<Map<String, Serializable>> all_node_data = 
          new ArrayList<Map<String, Serializable>>();
        for (int i = 0; i < my_message.get_AddNode_Message().getAddNodesList().size(); i++)
        {
          all_node_data.add(nodeman.node_stubs.get(nodes.get(i)).getData());
        }
        //Thread.sleep(1000);
        updateNodeManager(my_message, my_node_id);
        final AddNodeMessage new_mess = new AddNodeMessage(nodes, all_node_data);
        my_message.set_AddNode_Message(new_mess);
        my_message.setMy_originaldest_node(my_message.getMy_originalsource_node());
        my_message.setMy_originalsource_node(my_node_id);
        my_message.setMy_source_node(my_node_id);
        my_message.setAnswer_set(true);
        destNode = 
            nodeman.hashhelp.optimalNodeFind(my_node_id, my_message.getMy_originaldest_node());
        dest_str = nodeman.rmi_NodeFind(destNode);
        Logger.out(" inside the predecessor node " + my_node_id + 
            " transfering the data to the new node " + my_message.getMy_originaldest_node());
        if (destNode == my_message.getMy_originaldest_node())
        {
          my_message.setMy_dest_node(destNode);
          dest_str = destNode + "A";
        }
        final RemoteDistributedHashMap to_send = 
            (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_str);
        to_send.send(my_message);
      }
      else
      {
        if (!my_message.isAnswer_set() && my_message.getMy_originaldest_node() != my_node_id)
        {
          Logger.out(" add node message (tobe processed) received in node " + my_node_id + 
                     " forwarding it to node " + my_message.getMy_originaldest_node());
          //message not processed and on its way to the processing node.
          my_message.setMy_source_node(my_node_id);
          destNode = 
              nodeman.hashhelp.optimalNodeFind(my_node_id, my_message.getMy_originaldest_node());
          dest_str = nodeman.rmi_NodeFind(destNode);
          final RemoteDistributedHashMap to_send = 
            (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_str);
          to_send.send(my_message);
        }
        if (my_message.isAnswer_set() && my_message.getMy_originaldest_node() != my_node_id)
        {
          Logger.out(" add node message(processed) received in node " + my_node_id + 
                     " forwarding it to new node " + my_message.getMy_originaldest_node());
          my_message.setMy_source_node(my_node_id);
          destNode = 
              nodeman.hashhelp.optimalNodeFind(my_node_id, my_message.getMy_originaldest_node());
          dest_str = nodeman.rmi_NodeFind(destNode);
          if (destNode == my_message.getMy_originaldest_node())
          {
            my_message.setMy_dest_node(destNode);
            dest_str = destNode + "A";
          }
          final RemoteDistributedHashMap to_send = 
              (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_str);
          to_send.send(my_message);
        }
      }
    }
    catch (final NotBoundException  |  RemoteException e)
    {
      Logger.writeContent(nodeman.LOGGER_FILE_NAME, 
                          " not bound or problem in remote- exception in thread " + e);
      System.exit(1);
    }
    catch (final InterruptedException e)
    {
      Logger.writeContent(nodeman.LOGGER_FILE_NAME, " interrupted in thread " + e);
      System.exit(1);
    }
  }
  /**
   * process the message by creating virtual nodes, adding data and un binding the old node.
   */
  private void addVirtualNodes(final InterNodeMessage mess, final int my_node_id)
  {
    try
    {
      ArrayList<Integer> nodes_toAdd = new ArrayList<Integer>();
      nodes_toAdd = mess.get_AddNode_Message().getAddNodesList();
      Logger.out(" Number of new nodes to add " + nodes_toAdd.size());
      ArrayList<Map<String, Serializable>> all_data = 
          mess.get_AddNode_Message().getMy_node_data();
      for (int i = 1; i < nodes_toAdd.size(); i++)
      {
        nodeman.remoteArray[i] = new UnicastRemoteDistributedHashMap
        (nodes_toAdd.get(i), false, nodeman.actual_node_number,this.nodeman);
        Logger.out(" create/export stubs for new node " + nodes_toAdd.get(i));
        nodeman.node_stubs.put(nodes_toAdd.get(i), nodeman.remoteArray[i]);
        nodeman.remoteArray[i].setData(all_data.get(i));
        Logger.out(" transferring data to new node " + nodes_toAdd.get(i));
        nodeman.node_set.add(nodes_toAdd.get(i));
        final RemoteDistributedHashMap stub = (RemoteDistributedHashMap) 
            UnicastRemoteObject.exportObject(nodeman.remoteArray[i], 0);
        nodeman.messageChannelSetup(1);
        nodeman.QueueThreadStart(1);
        //set the channels and start thread.
        final String rmi_name = nodes_toAdd.get(i).toString() + "V";
        NodeManager.registry.rebind(rmi_name, stub);
        Logger.out(" binded new node " +  nodes_toAdd.get(i) + " to registry");
      }
    }
    catch (final RemoteException e)
    {
      Logger.err("a problem occurred:" + e);
      Logger.writeContent(nodeman.LOGGER_FILE_NAME, "a problem occurred : " + e);
      System.exit(1);
    }
  }
  /**
   * Update the NodeManager of the predecessor to reflect the changes of adding a node.
   */
  private void updateNodeManager(InterNodeMessage mess, int my_node_id)
  {
    ArrayList<Integer> allNodes = new ArrayList<Integer>();
    allNodes = mess.get_AddNode_Message().getAddNodesList();
    for (int i = 0; i <  allNodes.size(); i++)
    {
      //NodeManager.queueProcessors.get(allNodes.get(i)).setRun(false);
      //NodeManager.node_channel.remove(allNodes.get(i));
      //NodeManager.node_connections.remove(allNodes.get(i));
      nodeman.node_set.remove(allNodes.get(i));
      nodeman.node_stubs.remove(allNodes.get(i));
      nodeman.real_virtual_node_info.remove(allNodes.get(i));
    }
    Logger.out(" Node set in old node " + my_node_id + " are - ");
    final Iterator<Integer> itr = nodeman.node_set.iterator();
    while (itr.hasNext())
    {
      Logger.out(itr.next().toString());
    }
    
  }
}
 

